package com.googlecode.xcache.transcode.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.OutputStream;

import com.googlecode.xcache.transcode.TransCoder;
import com.schooner.MemCached.NativeHandler;

/**
 * A {@link TransCoder} implementation that supports the
 * {@link org.apache.thrift.protocol.TCompactProtocol TCompactProtocol} 
 * serialization and deserialization.
 * @author tielei
 *
 */
public class ThriftCompactTransCoder extends BaseTransCoder {
	/**
	 * Encode value using Thrift Compact protocol.
	 * values 
	 * @param value original value to encode.
	 * @return encoded value.
	 */
	@Override
	public Object encode(Object value) {
		if (NativeHandler.isHandled(value)) {
			//natively encoded by MemcachedClient; do not touch!
			return value;
		}
		try {
			Class<?> tBaseInterface = Class.forName("org.apache.thrift.TBase");
			if (tBaseInterface.isAssignableFrom(value.getClass())) {
				return thriftCompactWrite(value);
			}
		} catch (Exception e) {
			//thrift jar not exist, ignore, i.e., use underlying transcoding.
		}

		return value;
	}


	/**
	 * Decode value using Thrift Compact protocol.
	 * @param <T>
	 * @param expectedValueClass expected type of  value.
	 * @param value encoded value.
	 * @return original value.
	 */
	@Override
	public <T> Object customDecode(Class<T> expectedValueClass, Object value) {
		if (value == null) {
			//null value needs not to be further processing
			return null;
		}

		if (value instanceof byte[]) {
			try {
				Class<?> tBaseInterface = Class.forName("org.apache.thrift.TBase");
				if (tBaseInterface != null && tBaseInterface.isAssignableFrom(expectedValueClass)) {
					return thriftCompactRead(expectedValueClass, (byte[]) value);
				}
			} catch (Exception e) {
				//thrift jar not exist, ignore, i.e., use underlying transcoding.
			}
		}

		return value;
	}

	private byte[] thriftCompactWrite(Object tBaseModel) throws Exception {
		//Create TIOStreamTransport instance
		ByteArrayOutputStream os = new ByteArrayOutputStream();
		Class<?> tIOStreamTransportClass = Class.forName("org.apache.thrift.transport.TIOStreamTransport");
		Object tIOStreamTransportInstance = tIOStreamTransportClass.getConstructor(OutputStream.class).newInstance(os);

		//Create TCompactProtocol instance
		Class<?> tTransportClass = Class.forName("org.apache.thrift.transport.TTransport");
		Class<?> tCompactProtocolClass = Class.forName("org.apache.thrift.protocol.TCompactProtocol");
		Object tCompactProtocolInstance = tCompactProtocolClass.getConstructor(tTransportClass).newInstance(tIOStreamTransportInstance);

		//Call write on tBaseModel
		Class<?> tProtocolClass = Class.forName("org.apache.thrift.protocol.TProtocol");
		tBaseModel.getClass().getMethod("write", tProtocolClass).invoke(tBaseModel, tCompactProtocolInstance);
		return os.toByteArray();
	}

	private <T> T thriftCompactRead(Class<T> expectedValueClass, byte[] data) throws Exception {
		//Create TIOStreamTransport instance
		ByteArrayInputStream is = new ByteArrayInputStream(data);
		Class<?> tIOStreamTransportClass = Class.forName("org.apache.thrift.transport.TIOStreamTransport");
		Object tIOStreamTransportInstance = tIOStreamTransportClass.getConstructor(InputStream.class).newInstance(is);

		//Create TCompactProtocol instance
		Class<?> tTransportClass = Class.forName("org.apache.thrift.transport.TTransport");
		Class<?> tCompactProtocolClass = Class.forName("org.apache.thrift.protocol.TCompactProtocol");
		Object tCompactProtocolInstance = tCompactProtocolClass.getConstructor(tTransportClass).newInstance(tIOStreamTransportInstance);

		//Call read() to get tBaseModel
		T tBaseModel = expectedValueClass.newInstance();
		Class<?> tProtocolClass = Class.forName("org.apache.thrift.protocol.TProtocol");
		expectedValueClass.getMethod("read", tProtocolClass).invoke(tBaseModel, tCompactProtocolInstance);
		return tBaseModel;
	}

}
